package sample;


import org.apache.kafka.clients.producer.RecordMetadata;
import tech.mhuang.pacebox.kafka.admin.KafkaBuilder;
import tech.mhuang.pacebox.kafka.admin.KafkaFramework;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @package: com.hmtool
 * @author: mhuang
 * @Date: 2019/6/13 10:12
 * @Description:
 */
public class TestCreateProducer {

    public static void main(String[] args){
        //获取kafka组装器
        KafkaBuilder.Builder builder = KafkaBuilder.builder();
        //设置使用生产者
        builder.enableProducer(true);
        //构建生产者
        KafkaBuilder.ProducerBuilder producerBuilder = builder.createProducerBuilder();
        //生产者指定对应的服务
        producerBuilder.enable(true).servers("192.168.1.210:9092");
        //绑定生产者到组装器
        builder.bindProducer("test",producerBuilder.builder());
        //根据组装器构建到kafka
        KafkaFramework kafkaFramework = builder.builder();
        //启动kafka
        kafkaFramework.start();
        //用于测试发送
        while (true){
            Future<RecordMetadata> future =  kafkaFramework.getSuccessProducerMap().get("test").send("zhangsan","name","1",(meta, exception)->{
                System.out.println(meta);
                System.out.println(exception);
            });
            try {
                future.get();
                Thread.sleep(1000L);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
}
